package promote.brozen.reactor;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Operators;
import reactor.core.publisher.UnicastProcessor;
import reactor.util.concurrent.Queues;

import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @author Brozen
 * @date 2020/6/27 12:02 PM
 * @email brozen@qq.com
 */
public class FluxUnicastProcessor {

    public static void main(String[] args) throws InterruptedException {
        testBackPressure();
    }

    private static void testTransform() {
        UnicastProcessor<String> receiver = UnicastProcessor.create();
        receiver.transform(Operators.<String, String>lift((scannable, actual) -> new CoreSubscriber<String>() {

            @Override
            public void onSubscribe(Subscription s) {

            }

            @Override
            public void onNext(String o) {
                actual.onNext(o);
            }

            @Override
            public void onError(Throwable t) {
                actual.onError(t);
            }

            @Override
            public void onComplete() {
                actual.onComplete();
            }
        })).subscribe(System.out::println);
    }

    private static void testBackPressure() throws InterruptedException {
        UnicastProcessor<Integer> unicastProcessor = UnicastProcessor.create(Queues.<Integer>get(8).get());
        Flux<Integer> flux = unicastProcessor
                .map(e -> e)
                .doOnError(Throwable::printStackTrace);

        flux.subscribe(System.out::println);

        IntStream.range(0, 20).forEach(i -> {
            System.out.println("send:" + i);
            unicastProcessor.onNext(i);
        });

        TimeUnit.SECONDS.sleep(1);

        unicastProcessor.onComplete();
    }
}
